package com.niit.adversStat

import com.niit.adversStat.dao._
import com.niit.adversStat.model._
import com.niit.commons.conf.ConfigurationManager
import com.niit.commons.constant.Constants
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.lang

object Test {
  def main(args: Array[String]): Unit = {

    val sparkconf = new SparkConf().setAppName("adverStat").setMaster("local[*]")
    val sparkSession = SparkSession.builder().config(sparkconf).enableHiveSupport().getOrCreate()
    val streamingContext = new StreamingContext(sparkSession.sparkContext, Seconds(5))

    val kafka_brokers = "192.168.10.102:9092"
    val kafka_topics = ConfigurationManager.config.getString(Constants.KAFKA_TOPICS)

    val kafkaParam = Map(
      "bootstrap.servers" -> kafka_brokers,
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "AdRealTimeLog",
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: lang.Boolean)
    )

    val adRealTimeDStream = KafkaUtils.createDirectStream(streamingContext,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Array(kafka_topics), kafkaParam)
    )

    val adRealTimeValueDStream: DStream[String] = adRealTimeDStream.map(item => item.value())
//    adRealTimeValueDStream.foreachRDD { rdd =>
//      rdd.foreach { record =>
//        println(s"Record: $record")
//      }
//    }
    val filterBlackListRDD: DStream[String] = adRealTimeValueDStream.transform(
      logRDD => {
        val blacklists: Array[AdBlacklist] = AdBlacklistDAO.findAll()
        val blackListUserID: Array[Long] = blacklists.map(item => item.userid)
        val filterUserIdLogRDD: RDD[String] = logRDD.filter {
          case log =>
            val userid = log.split(" ")(3)
            !blackListUserID.contains(userid.toLong)
        }
        filterUserIdLogRDD.foreach(println)
        filterUserIdLogRDD
      }
    )

//    streamingContext.checkpoint("./spark-streaming")
//    filterBlackListRDD.checkpoint(Duration(10000))

    // 打印 filterBlackListRDD 到控制台
    filterBlackListRDD.foreachRDD { rdd =>
      rdd.foreach { record =>
        println(record+"1111111111111111")
      }
    }

    streamingContext.start()
    streamingContext.awaitTermination()
  }
}
